package org
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.desc
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
object sparkYun2_SQL3 {
  def main(args: Array[String]): Unit = {
    //创建spark运行环境
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
    //表结构读取电影用户数据
    val schemaUser = StructType(Seq(
      StructField("id", IntegerType),
      StructField("gender", StringType),
      StructField("age",IntegerType),
      StructField("occupation", IntegerType),
      StructField("location",StringType)
    ))
    val user = spark.read.option("sep","::").schema(schemaUser)
      .csv("src/main/resources/users.dat")
    //user.show(5)

    //1.查询 替换 where/filter
   // user.where("gender = 'F' and age =18").show(3)
    //user.filter("gender = 'F' and age =18").show(3)

//    spark.udf.register("replace",(x:String)=>{
//
//      x match{
//        case "M"=>0
//        case "F"=>1
//      }
//    })
//    user.selectExpr("id","replace(gender) as sexual","age").show(3)
//    user.select("id","age","location").show(3)
//
//    //2.排序
//    user.orderBy(desc("id")).show(5)
//    user.sort(-user("id")).show(5)
//    //分组
//    user.groupBy("gender").count().show()
    //连接
    //练习 5.练习：将users用户表跟rating评分表还有movies电影表连接起来，查找年龄18岁的女生评分为5 分的所有电影名字
// 读取评分数据
val schemaRating = StructType(Seq(
  StructField("userId", IntegerType, nullable = false),
  StructField("movieId", IntegerType, nullable = false),
  StructField("rating", IntegerType),
  StructField("timestamp", StringType)
))
    val rating = spark.read.option("sep", "::").schema(schemaRating)
      .csv("src/main/resources/ratings.dat")

    // 读取电影数据
    val schemaMovie = StructType(Seq(
      StructField("movieId", IntegerType, nullable = false),
      StructField("title", StringType),
      StructField("genres", StringType)
    ))
    val movie = spark.read.option("sep", "::").schema(schemaMovie)
      .csv("src/main/resources/movies.dat")

    // 查询18岁女生评分电影为5分的所有电影名
    val result = user
      .where("gender = 'F' and age = '18'")
      .join(rating, user("id") === rating("userId"))
      .where("rating = 5")
      .join(movie, rating("movieId") === movie("movieId"))
      .select("title")
      .distinct()

    result.show()
    sc.stop()
}}
